package com.alibaba.datax.plugin.reader.mysqlreader.manager;

import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.element.StringColumn;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.rdbms.util.DBUtil;
import com.alibaba.druid.pool.DruidDataSourceFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Map;

public class AddCustNoPreHandler implements PreHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(AddCustNoPreHandler.class);

    private DataSource dataSource;
    private String querySql;
    private Map<String, Object> pkMapping;

    public AddCustNoPreHandler(Configuration configuration) {
        try {
            Map<String, Object> dataSourceConfig = configuration.getMap("datasourceConfig");
            dataSource = DruidDataSourceFactory.createDataSource(dataSourceConfig);

            this.querySql = configuration.getString("querySql");
            this.pkMapping = configuration.getMap("addCustNoPreHandlerPkMapping");
        } catch (Exception e) {
            LOGGER.error("创建前置处理器异常", e);
            throw new RuntimeException(e);
        }

    }

    @Override
    public void preHandle(Record record) {
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        try {
            connection = dataSource.getConnection();
            preparedStatement = connection.prepareStatement(querySql);
            int paramIndex = 1;
            for (Map.Entry<String, Object> entry : pkMapping.entrySet()) {
                Map<String, Object> value = (Map<String, Object>) entry.getValue();
                JDBCUtil.setPStmt((Integer) value.get("psmtType"), preparedStatement, record.getColumn((Integer) value.get("index")).getRawData(), paramIndex++);
            }
            resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                record.addColumn(new StringColumn(resultSet.getString(1)));
            }
        } catch (Exception e) {
            LOGGER.error("前置处理器处理过程异常：record：{}", recordToStr(record));
        } finally {
            DBUtil.closeDBResources(resultSet, preparedStatement, connection);
        }
    }

    private String recordToStr(Record record) {
        StringBuilder stringBuilder = new StringBuilder();
        for (int i = 0; i < record.getColumnNumber(); i++) {
            if (i != 0) {
                stringBuilder.append(",");
            }
            stringBuilder.append(record.getColumn(i).getRawData());
        }
        return stringBuilder.toString();
    }
}